package com.kl.example.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/**
 * 主要在消费到消息或在提交消费位移时进行一些定制化的操作
 * 消费者拦截器示例：实现一个简单的消息TTL（过期时间）的功能，
 * 使用消息的 timestamp 字段来判断消息是否过期，如果消息的时间戳与当前的时间戳相差超过10秒则判定为过期，
 * 那么这条消息也就被过滤而不投递给其他消费者
 */
public class ConsumerInterceptorTTLExample implements ConsumerInterceptor<String, String> {

    private final static long EXPIRE_INTERVAL = 10 * 1000;

    /**
     * 在 poll 方法返回之前调用
     * 可以用来修改返回的消息内容，或者按照某种过滤规则过滤消息
     */
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // 当前时间
        long now = System.currentTimeMillis();
        // 分区对应的消息集合
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>();
        // 遍历当前所有的分区
        Set<TopicPartition> partitions = records.partitions();
        for (TopicPartition partition : partitions) {
            // 获取指定分区中的消息集合
            List<ConsumerRecord<String, String>> tpRecords = records.records(partition);
            // 满足条件后新的消息集合
            List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
            // 过滤原消息集合
            for (ConsumerRecord<String, String> tpRecord : tpRecords) {
                if (now - tpRecord.timestamp() < EXPIRE_INTERVAL) {
                    newTpRecords.add(tpRecord);
                }
            }
            // 有过滤出满足条件的消息,将新的分区->映射消息列表的和保存到总的集合列表中
            if (!newTpRecords.isEmpty()) {
                newRecords.put(partition, newTpRecords);
            }

        }
        return new ConsumerRecords<>(newRecords);
    }

    /**
     * 在提交消费位移之后调用
     * 可以用来记录跟踪所提交的位移信息，比如当消费者使用 commitSync 的无参方法时，不知道提交的消费位移的具体细节，
     * 但使用该方法就可以获取到
     */
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 遍历 k-v
        offsets.forEach((tp, offset) -> {
            System.out.println(tp + ":" + offset.offset());
        });
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
